// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_ROW_MUTATOR_HPP_
#define _XBASE_ROW_MUTATOR_HPP_

#include "async_module_base.hpp"
#include "sync_handler.hpp"
#include "location_cache.hpp"
#include "row_mutator_on_tablet.hpp"
#include "row_mutation.hpp"

namespace xbase
{

	class row_mutator
		: public async_module_base<io_service_pool::user_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::user_level, io_service_pool::xbase_level> base_type;
		typedef row_mutator this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef session_manager::session_ptr session_ptr;
		typedef row_mutator_on_tablet::ptr_type row_mutator_on_tablet_ptr;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(const error_code&)> handler_type;
		typedef basic_sync_handler<> sync_handler_type;

		typedef protocol::packet_type<protocol::WRITE_REQ>::type write_req;
		typedef protocol::packet_type<protocol::WRITE_REQ>::ptr_type write_req_ptr;

	public:
		explicit
		row_mutator(io_service_pool &pool,
			    session_manager &manager,
			    location_cache &cache)
			: base_type(pool, "row_mutator")
			, m_io_service_pool(pool)
			, m_options(pool.options())
			, m_session_manager(manager)
			, m_location_cache(cache)
			, m_retry_times(0)
			, m_tablet_info_invalid(false) {}

		// 将修改操作应用到指定表
		//
		// 用户未指定的时间戳将在此时取当前系统时间来代替
		//
		// 注意：尽量做到提交操作不修改自己！
		template<typename Handler>
		inline
		void async_apply(const std::string &table_name,
				 const row_mutation_data_ptr &mutation,
				 uint64_t timestamp,
				 Handler handler) {
			// 不能写空行！
			if(mutation->row_name().empty())
			{
				this->get_upper_level_service().dispatch(
					boost::bind(handler, error::bad_row_name)
					);
				return;
			}

			write_req_ptr req_ptr(new write_req());
			protocol::write_request_v2 &body = req_ptr->body();
			//body.m_tablet_name = tablet_name;
			body.m_mutation_data = mutation;
			body.m_timestamp = timestamp;
			//mutation.fill_record_vector(body.m_records, timestamp);

			m_table_name = table_name;
			m_row_name = mutation->row_name();
			m_req_ptr = req_ptr;
			m_handler = handler;

			m_retry_times = 0;
			m_tablet_info_invalid = false;
			
			// 初始化为0
			m_meta_data.meta_got_time = 0;

			this->get_lower_level_service().dispatch(
				boost::bind(&this_type::on_async_apply, this_ptr())
				);
		}

		template<typename Handler>
		inline
		void async_apply(const std::string &table_name,
				 const row_mutation_data_ptr &mutation,
				 Handler handler) {
			async_apply(table_name,
				    mutation,
				    cur_micro_seconds(),
				    handler);
		}

		error_code apply(const std::string &table_name,
				 const row_mutation_data_ptr &mutation,
				 uint64_t timestamp) {
			sync_handler_type sh;
			async_apply(table_name,
				    mutation,
				    timestamp,
				    sh.make_handler());
			sh.wait();
			return sh.get_error_code();
		}
		
		error_code apply(const std::string &table_name,
				 const row_mutation_data_ptr &mutation) {
			sync_handler_type sh;
			async_apply(table_name,
				    mutation,
				    sh.make_handler());
			sh.wait();
			return sh.get_error_code();
		}

	protected:
		void on_async_apply();

		void on_got_tablet(location_cache_value meta, const error_code &ec);

		void on_session_connected(session_ptr session, const error_code &ec);

		void on_complete(const error_code &ec);

		void on_retry(const error_code &ec);

		void on_error(const error_code &ec);

		void dispatch(const error_code &ec);

		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		io_service_pool &m_io_service_pool;
		const client_options &m_options;
		session_manager &m_session_manager;
		location_cache &m_location_cache;

		std::string m_table_name;
		std::string m_row_name;
		write_req_ptr m_req_ptr;
		handler_type m_handler;

		boost::shared_ptr<boost::asio::deadline_timer> m_timer;
		std::size_t m_retry_times;
		bool m_tablet_info_invalid;

		location_cache_value m_meta_data;
		row_mutator_on_tablet_ptr m_tablet_mutator;
	};

	typedef row_mutator::ptr_type row_mutator_ptr;

	template<typename Handler>
	inline
	void async_apply(io_service_pool &pool,
			 session_manager &manager,
			 location_cache &cache,
			 const std::string &table_name,
			 const row_mutation_data_ptr &mutation,
			 uint64_t timestamp,
			 Handler handler) {
		row_mutator_ptr mutator(new_async_module(row_mutator, pool, manager, cache));
		mutator->async_apply(table_name, mutation,
				     timestamp, handler);
	}

	template<typename Handler>
	inline
	void async_apply(io_service_pool &pool,
			 session_manager &manager,
			 location_cache &cache,
			 const std::string &table_name,
			 const row_mutation_data_ptr &mutation,
			 Handler handler) {
		row_mutator_ptr mutator(new_async_module(row_mutator, pool, manager, cache));
		mutator->async_apply(table_name, mutation,
				     handler);
	}

	template<typename Handler>
	inline
	void async_apply(io_service_pool &pool,
			 session_manager &manager,
			 location_cache &cache,
			 const row_mutation &mutation,
			 uint64_t timestamp,
			 Handler handler) {
		row_mutator_ptr mutator(new_async_module(row_mutator, pool, manager, cache));
		mutator->async_apply(mutation.table_name(), mutation.data_ptr(),
				     timestamp, handler);
	}

	template<typename Handler>
	inline
	void async_apply(io_service_pool &pool,
			 session_manager &manager,
			 location_cache &cache,
			 const row_mutation &mutation,
			 Handler handler) {
		row_mutator_ptr mutator(new_async_module(row_mutator, pool, manager, cache));
		mutator->async_apply(mutation.table_name(), mutation.data_ptr(),
				     handler);
	}

} // namespace xbase

#endif	// _XBASE_ROW_MUTATOR_HPP_
